package com.dongdongshop.mq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.dongdongshop.service.TbUserScoreAccountService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
//消费者需要用注解配置topic主题 和消费者的组
@RocketMQMessageListener(topic = "tx-order-score", consumerGroup = "ggw")
//消费者需要继承接口RocketMQListener<>可以通过泛型接受消费消息
//ack重试机制要继承RocketMQPushConsumerLifecycleListener
public class Consume implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private TbUserScoreAccountService tbUserScoreAccountService;


    //继承RocketMQListener的方法
    @Override
    public void onMessage(String s) {
        //继承RocketMQPushConsumerLifecycleListener的接口后这个方法就作废了
        System.out.println("接收到信息 : " + s);
    }

    //继承RocketMQPushConsumerLifecycleListener后的方法
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        //通过方法创建内部类内部类中写业务
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //list就是消费消息
                //单条消息直接下表0获取对象,可以通过对象获取更多属性
                MessageExt messageExt = list.get(0);
                //ctrl+alt+t  生成try
                try {
                    //new string接受消息内容 如果不转string会是一个二进制字符串
                    String msg = new String(messageExt.getBody());
                    JSONObject jsonObject = JSONUtil.parseObj(msg);
                    System.out.println(jsonObject.toString());
                    String userid1 = jsonObject.get("userid").toString();
                    Long userid = Long.valueOf(userid1);
                    String out_trade_no1 = jsonObject.get("out_trade_no").toString();
                    Long out_trade_no = Long.valueOf(out_trade_no1);
                    String receipt_amount1 = jsonObject.get("receipt_amount").toString();
                    int i = receipt_amount1.indexOf(".");
                    String substring = receipt_amount1.substring(0, i);
                    Long receipt_amount = Long.valueOf(substring);
                    tbUserScoreAccountService.add(userid,out_trade_no , receipt_amount);
                    System.out.println("积分添加成功");
                } catch (Exception e) {
                    e.printStackTrace();
                    //获取消息重试次数
                    int reconsumeTimes = messageExt.getReconsumeTimes();
                    if (reconsumeTimes >= 3) {//如果重试次数大于3,没必要继续尝试返回成功,并且将消息放入死信队列
                        System.out.println("放入死信队列");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    System.out.println("ACK消息接受失败,请稍后重试");
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;//发生异常返回重新消费
                }
                //消费成功返回success
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
